package edu.zju.gis.dbfg.model.datasource;


import com.google.gson.Gson;
import com.google.gson.internal.LinkedTreeMap;
import edu.zju.gis.dbfg.model.PyramidConfig;
import edu.zju.gis.dbfg.model.base.Constant;
import edu.zju.gis.dbfg.model.util.GeoItem;
import edu.zju.gis.dbfg.model.util.TileID;
import edu.zju.gis.dbfg.model.util.TileUtil;
import edu.zju.gis.dbfg.model.util.ZLevelInfo;
import org.apache.http.HttpHost;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.elasticsearch.action.search.*;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.geo.ShapeRelation;
import org.elasticsearch.common.geo.builders.EnvelopeBuilder;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.index.query.GeoShapeQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.geotools.geometry.jts.JTS;
import org.geotools.geometry.jts.ReferencedEnvelope;
import org.geotools.referencing.CRS;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Envelope;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.GeometryFactory;
import org.locationtech.jts.io.WKTReader;
import org.opengis.feature.simple.SimpleFeature;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.io.File;
import java.util.*;
import java.util.stream.Collectors;

import static edu.zju.gis.dbfg.model.base.Constant.*;
import static org.elasticsearch.index.query.QueryBuilders.geoShapeQuery;

/**
 * 读取数据并转化为 JavaRDD<GeoItem> 或 JavaPairRDD<String, GeoItem>
 * 其中 String 为所格网ID
 */
public class DataReaderRDD {

    private static final Logger logger = LoggerFactory.getLogger(DataReaderRDD.class);

    public static JavaRDD<GeoItem> readToRDD(String type, JavaSparkContext jsc, String file, String idIndex, Map<String, Object> attrs, MathTransform transform) {

        switch (type) {
            case "WKT":
                Map<String, Integer> attrsInt = null;
                if(null != attrs) {
                    attrsInt = new HashMap<>();
                    for (String key : attrs.keySet()) {
                        Integer v = Integer.valueOf((String) attrs.get(key));
                        attrsInt.put(key, v);
                    }
                }
                JavaRDD<GeoItem> gi=readFromFile(jsc, file, Integer.valueOf(idIndex), attrsInt, transform);
                return gi;
            case "SHP":
                Map<String, String> attrsStr = null;
                if(null != attrs) {
                    attrsStr = new HashMap<>();
                    for (String key : attrs.keySet()) {
                        attrsStr.put(key, (String) attrs.get(key));
                    }
                }
                return readFromShp(jsc, file, attrsStr, transform);
            default:
                logger.error("Unsupport RDD Read Type");
                return null;
        }
    }

    public static JavaPairRDD<String, GeoItem> readToPairRDD(String type, JavaSparkContext jsc, String source, Map<String, Object> attrs, List<TileID> tiles, PyramidConfig pc) throws Exception {
        switch (type) {
            case "ES":
                Map<String, String> attrsStr = null;
                if(null != attrs){
                    attrsStr = new HashMap<>();
                    for (String key : attrs.keySet()) {
                        attrsStr.put(key, (String) attrs.get(key));
                    }
                }
                return readFromES(jsc, source, attrsStr, tiles, pc);
            default:
                logger.error("Unsupport Pair RDD Read Type");
                return null;
        }
    }

    public static JavaRDD<GeoItem> readToRDD2(String type, JavaSparkContext jsc, String source, Map<String, Object> attrs, List<String> filters, String filterName, MathTransform transform) throws Exception {
        switch (type) {
            case "ES":
                Map<String, String> attrsStr = null;
                if(attrs != null) {
                    attrsStr = new HashMap<>();
                    for (String key : attrs.keySet()) {
                        attrsStr.put(key, (String) attrs.get(key));
                    }
                }
                //如果需要按字段过滤
                if(filterName != null){
                    return readFromES2(jsc, source, attrsStr, filters, filterName, transform);
                }
                else {
                    return readFromES3(jsc, source, attrsStr, transform);
                }
            default:
                logger.error("Unsupport Pair RDD Read Type");
                return null;
        }
    }


    /**
     * 从wkt文件中读取GeoItem，默认最后一列为wkt
     * TODO geoitem 的 prop 中的 value 均为 string
     *
     * @param jsc
     * @param file
     * @param attrs
     * @param transform
     * @return
     */
    private static JavaRDD<GeoItem> readFromFile(JavaSparkContext jsc, String file, Integer idIndex, Map<String, Integer> attrs, MathTransform transform) {
        JavaRDD<GeoItem> result = jsc.textFile(file).map(new Function<String, GeoItem>() {
            @Override
            public GeoItem call(String s) throws Exception {
                GeoItem gi = new GeoItem();
                WKTReader reader = new WKTReader();
                String[] fields = s.split("\t");
                String wkt = fields[0].replace("\"","");
                Geometry geometry = reader.read(wkt);
                if (transform != null) {
                    geometry = JTS.transform(geometry, transform);
                }
                gi.setFid(fields[idIndex]);
                gi.setGeometry(geometry);
                Map<String, Object> props = new HashMap<>();
                if(null == attrs){
                    for(int i = 0; i < fields.length-1; i++){
                        props.put(String.valueOf(i), fields[i]);
                    }
                }
                else {
                    for (String key : attrs.keySet()) {
                        props.put(key, fields[attrs.get(key)]);
                    }
                }
                gi.setProp(props);
                return gi;
            }
        });
        return result;
    }

    private static JavaRDD<GeoItem> readFromShp(JavaSparkContext jsc, String path, Map<String, String> attrs, MathTransform transform) {

        JavaRDD<GeoItem> result = null;

        List<String> filesPath = new ArrayList<>();
        File file = new File(path);
        if (file.isDirectory()) {
            File[] files = file.listFiles();
            for (File f : files) {
                if (f.getAbsolutePath().endsWith(".shp")) {
                    filesPath.add(f.getAbsolutePath());
                }
            }
        } else {
            filesPath.add(file.getAbsolutePath());
        }
        result = jsc.parallelize(filesPath).flatMap(new FlatMapFunction<String, GeoItem>() {
            @Override
            public Iterator<GeoItem> call(String s) throws Exception {
                ShpDataReader reader = new ShpDataReader(s);
                reader.init();
                List<GeoItem> result = new ArrayList<>();
                SimpleFeature feature = reader.nextFeature();
                while (feature != null) {
                    GeoItem gi = new GeoItem();
                    Geometry geometry = (Geometry) feature.getDefaultGeometry();
                    if (transform != null) {
                        System.out.println("===="+geometry.toString());
                        geometry = JTS.transform(geometry, transform);
                        System.out.println("------"+geometry.toString());
                    }
                    gi.setGeometry(geometry);
                    gi.setFid(feature.getID());

                    Map<String, Object> props = new HashMap<>();

                    if(null == attrs){
                        List<String> fieldName = reader.readFieldName();
                        for(int i = 0; i < fieldName.size(); i++){
                            props.put(fieldName.get(i), feature.getAttribute(fieldName.get(i)));
                        }
                    }
                    else {
                        for (String key : attrs.keySet()) {
                            props.put(key, feature.getAttribute(attrs.get(key)));
                        }
                    }

                    gi.setProp(props);

                    result.add(gi);
                    feature = reader.nextFeature();
                }
                reader.close();
                return result.iterator();
            }
        });

        return result;
    }

    /**
     * source 以 es:// 开头
     * 支持以逗号分隔的多个 es source
     *
     * @param jsc
     * @param source
     * @param attrs
     * @param tiles
     * @return
     */
    public static JavaPairRDD<String, GeoItem> readFromES(JavaSparkContext jsc, String source, Map<String, String> attrs, List<TileID> tiles, PyramidConfig pc) throws Exception {
        List<String> sources = Arrays.stream(source.split(",")).map(String::trim).map(x -> x.replace("es://", "")).collect(Collectors.toList());
        List<Tuple2<String, TileID>> searchQ = new ArrayList<>();
        CoordinateReferenceSystem crs = pc.getCrs();
        CoordinateReferenceSystem crs_4326 = Constant.CRS_4326();
        MathTransform tstransform = null;
        if (!crs.equals(crs_4326)) {
            tstransform = CRS.findMathTransform(crs, crs_4326);
        }
        for (String src : sources) {
            for (TileID tile : tiles) {
                searchQ.add(new Tuple2<>(src, tile));
            }
        }
        Broadcast<MathTransform> tstransformBroad = jsc.broadcast(tstransform);

        JavaPairRDD<String, GeoItem> resultRDD = jsc.parallelizePairs(searchQ, (searchQ.size() / 64) + 1).flatMapToPair(new PairFlatMapFunction<Tuple2<String, TileID>, String, GeoItem>() {
            @Override
            public Iterator<Tuple2<String, GeoItem>> call(Tuple2<String, TileID> input) throws Exception {
                String[] ips = ES_IP.split(",");
                HttpHost[] hosts = new HttpHost[ips.length];
                for (int m = 0; m < ips.length; m++) {
                    hosts[m] = new HttpHost(ips[m].trim(), ES_IP_PORT);
                }
                RestHighLevelClient client = new RestHighLevelClient(
                        RestClient.builder(hosts));
                List<Tuple2<String, GeoItem>> result = new ArrayList<>();
                    List<Tuple2<String, GeoItem>> features = new ArrayList<>();
                    List<Tuple2<String, String>> tuples = new ArrayList<>();
                    Tuple2<String, TileID> in = input;
                    Envelope tile = TileUtil.createTileBox(in._2, pc);
                    if (tstransformBroad.getValue() != null) {
                        tile = JTS.transform(tile, tstransformBroad.getValue());
                    }
                    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
                    logger.info("======start struct shape=======");
                    EnvelopeBuilder envelopeBuilder = new EnvelopeBuilder(new Coordinate(tile.getMinX(), tile.getMaxY()), new Coordinate(tile.getMaxX(), tile.getMinY()));
                    GeoShapeQueryBuilder qb = geoShapeQuery(
                            "GEOM",//zy_modify
                            envelopeBuilder);
                    logger.info("======end struct shape=======");

                    qb.relation(ShapeRelation.INTERSECTS);
                    sourceBuilder.query(qb);
                    sourceBuilder.size(ES_SEARCH_SCROLL_SIZE);
                    // first search request to get one batch of data
                    SearchRequest searchRequest = new SearchRequest();
                    searchRequest.indices(source);
                    searchRequest.source(sourceBuilder);
                    searchRequest.scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
                    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                    SearchHits hits = searchResponse.getHits();
                    SearchHit[] searchHits = hits.getHits();
                    for (SearchHit hit : searchHits) {
                        tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                    }
                    // scroll request to get all left relevant data
                    String scrollId = searchResponse.getScrollId();
                    final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
                    while (searchHits != null && searchHits.length > 0) {
                        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                        scrollRequest.scroll(scroll);
                        searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                        scrollId = searchResponse.getScrollId();
                        searchHits = searchResponse.getHits().getHits();
                        for (SearchHit hit : searchHits) {
                            tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                        }
                    }
                    // clear scroll id
                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                    clearScrollRequest.addScrollId(scrollId);
                    client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
                    /**
                     * 对于获取到的数据,转为GeoItem
                     */


                    for (Tuple2<String, String> f : tuples) {
                        Gson gson = new Gson();
                        Map<String, Object> inMap = gson.fromJson(f._2, new HashMap<String, Object>().getClass());
                        inMap.remove("id");
                        if (!inMap.containsKey("GEOM")) {
                            logger.error("Unvalid geometry item for id " + f._1 + "with source " + source + ", must have field named with 'shape'");
                            continue;
                        }
                        Map<String, Object> geoMap = (LinkedTreeMap) inMap.get("GEOM");
                        String type = (String) geoMap.get("type");
                        ArrayList t = ((ArrayList) geoMap.get("coordinates"));
                        GeometryFactory gf = new GeometryFactory();
                        if (type.equals("Polygon") || type.equals("MultiPolygon")){
                            for (int i = 0; i < t.size(); i++) {
                                List<List<ArrayList>> coordinates = new ArrayList<>();
                                ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
                                if (!type.contains("Multi")) {
                                    coordinates.add(m);
                                } else {
                                    for (int x = 0; x < m.size(); x++) {
                                        coordinates.add(m.get(x));
                                    }
                                }
                                for (List<ArrayList> coordinate : coordinates) {
                                    Geometry geometry;
                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
                                    geometry = gf.createPolygon(cs);

                                Map<String, Object> props = new HashMap<>();
                                Double yarea=geometry.getArea();
                                props.put("YTBMJ",yarea);
                                Double ylength=geometry.getLength();
                                props.put("YLENGTH",ylength);
                                if(null == attrs){  //如果没有声明attrs，则默认读取全部属性
                                    props = inMap;
                                    props.remove("GEOM");
                                }
                                else { //如果声明了attrs，则根据attrs中的key来获取属性
                                    for (String key : attrs.keySet()) {
                                        props.put(key, inMap.get(attrs.get(key)));
                                    }
                                }
                                if (tstransformBroad.getValue() != null) {
                                    geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
                                }

                                GeoItem gi = new GeoItem();
                                gi.setFid(f._1);
                                gi.setProp(props);
                                gi.setGeometry(geometry);

                                features.add(new Tuple2<>(in._2.toString(), gi));
                            }


                            }
                        }else if (type.equals("LineString") || type.equals("MultiLineString")) {
                            for (int i = 0; i < t.size(); i++) {
                                List<ArrayList> coordinates = new ArrayList<>();
                                ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
                                coordinates.add(m);

                                for (List<ArrayList> coordinate : coordinates) {
                                    Geometry geometry;
                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
                                    geometry = gf.createLineString(cs);

                                    Map<String, Object> props = new HashMap<>();

                                    Double ylength=geometry.getLength();
                                    props.put("YLENGTH",ylength);
                                    if(null == attrs){  //如果没有声明attrs，则默认读取全部属性
                                        props = inMap;
                                        props.remove("GEOM");
                                    }
                                    else { //如果声明了attrs，则根据attrs中的key来获取属性
                                        for (String key : attrs.keySet()) {
                                            props.put(key, inMap.get(attrs.get(key)));
                                        }
                                    }
                                    if (tstransformBroad.getValue() != null) {
                                        geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
                                    }

                                    GeoItem gi = new GeoItem();
                                    gi.setFid(f._1);
                                    gi.setProp(props);
                                    gi.setGeometry(geometry);

                                    features.add(new Tuple2<>(in._2.toString(), gi));
                                }


                            }



                        }else {
                            Geometry geometry;
                            Coordinate c = new Coordinate(Double.valueOf (t.get(0).toString()), Double.valueOf (t.get(1).toString()));
                            geometry = gf.createPoint(c);
                            Map<String, Object> props = new HashMap<>();
//                            Double yarea=geometry.getArea();
//                            props.put("YTBMJ",yarea);
//                            Double ylength=geometry.getLength();
//                            props.put("YLENGTH",ylength);
                            if(null == attrs){  //如果没有声明attrs，则默认读取全部属性
                                props = inMap;
                                props.remove("GEOM");
                            }
                            else { //如果声明了attrs，则根据attrs中的key来获取属性
                                for (String key : attrs.keySet()) {
                                    props.put(key, inMap.get(attrs.get(key)));
                                }
                            }
                            if (tstransformBroad.getValue() != null) {
                                geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
                            }

                            GeoItem gi = new GeoItem();
                            gi.setFid(f._1);
                            gi.setProp(props);
                            gi.setGeometry(geometry);

                            features.add(new Tuple2<>(in._2.toString(), gi));

                        }




//                        for (int i = 0; i < t.size(); i++) {
//                            List<List<ArrayList>> coordinates = new ArrayList<>();
//                            ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
//                            if (!type.contains("Multi")) {
//                                coordinates.add(m);
//                            } else {
//                                for (int x = 0; x < m.size(); x++) {
//                                    coordinates.add(m.get(x));
//                                }
//                            }
//                            for (List<ArrayList> coordinate : coordinates) {
//                                Geometry geometry;
//                                if (type.equals("Polygon") || type.equals("MultiPolygon")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createPolygon(cs);
//                                } else if (type.equals("LineString") || type.equals("MultiLineString")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createLineString(cs);
//                                } else {
//                                    // type == Point
//                                    Coordinate c = new Coordinate(Double.valueOf (coordinate.get(0).get(0).toString()), Double.valueOf (coordinate.get(0).get(1).toString()));
//                                    geometry = gf.createPoint(c);
//                                }
//
//                                Map<String, Object> props = new HashMap<>();
//                                Double yarea=geometry.getArea();
//                                props.put("YTBMJ",yarea);
//                                Double ylength=geometry.getLength();
//                                props.put("YLENGTH",ylength);
//                                if(null == attrs){  //如果没有声明attrs，则默认读取全部属性
//                                    props = inMap;
//                                    props.remove("GEOM");
//                                }
//                                else { //如果声明了attrs，则根据attrs中的key来获取属性
//                                    for (String key : attrs.keySet()) {
//                                        props.put(key, inMap.get(attrs.get(key)));
//                                    }
//                                }
//                                if (tstransformBroad.getValue() != null) {
//                                    geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
//                                }
//
//                                GeoItem gi = new GeoItem();
//                                gi.setFid(f._1);
//                                gi.setProp(props);
//                                gi.setGeometry(geometry);
//
//                                features.add(new Tuple2<>(in._2.toString(), gi));
//                            }
//                        }
                    }

                    result.addAll(features);

                client.close();
                return result.iterator();
            }
        });


//        JavaPairRDD<String, GeoItem> resultRDD = jsc.parallelizePairs(searchQ, (searchQ.size() / 64) + 1).mapPartitionsToPair(new PairFlatMapFunction<Iterator<Tuple2<String, TileID>>, String, GeoItem>() {
//            @Override
//            public Iterator<Tuple2<String, GeoItem>> call(Iterator<Tuple2<String, TileID>> input) throws Exception {
//                String[] ips = ES_IP.split(",");
//                HttpHost[] hosts = new HttpHost[ips.length];
//                for (int m = 0; m < ips.length; m++) {
//                    hosts[m] = new HttpHost(ips[m].trim(), ES_IP_PORT);
//                }
//
//                RestHighLevelClient client = new RestHighLevelClient(
//                        RestClient.builder(hosts));
//
//                List<Tuple2<String, GeoItem>> result = new ArrayList<>();
//
//                while (input.hasNext()) {
//
//                    List<Tuple2<String, GeoItem>> features = new ArrayList<>();
//
//                    List<Tuple2<String, String>> tuples = new ArrayList<>();
//
//                    Tuple2<String, TileID> in = input.next();
//
//                    Envelope tile = TileUtil.createTileBox(in._2, pc);
//                    if (tstransformBroad.getValue() != null) {
//                        tile = JTS.transform(tile, tstransformBroad.getValue());
//                    }
//
//                    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//
//                    logger.info("======start struct shape=======");
//                    EnvelopeBuilder envelopeBuilder = new EnvelopeBuilder(new Coordinate(tile.getMinX(), tile.getMaxY()), new Coordinate(tile.getMaxX(), tile.getMinY()));
//                    GeoShapeQueryBuilder qb = geoShapeQuery(
//                            "shape",
//                            envelopeBuilder);
//                    logger.info("======end struct shape=======");
//
//                    qb.relation(ShapeRelation.INTERSECTS);
//                    sourceBuilder.query(qb);
//                    sourceBuilder.size(ES_SEARCH_SCROLL_SIZE);
//                    // first search request to get one batch of data
//                    SearchRequest searchRequest = new SearchRequest();
//                    searchRequest.indices(source);
//                    searchRequest.source(sourceBuilder);
//                    searchRequest.scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
//                    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
//                    SearchHits hits = searchResponse.getHits();
//                    SearchHit[] searchHits = hits.getHits();
//
//                    for (SearchHit hit : searchHits) {
//                        tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
//                    }
//
//                    // scroll request to get all left relevant data
//                    String scrollId = searchResponse.getScrollId();
//                    final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
//                    while (searchHits != null && searchHits.length > 0) {
//                        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
//                        scrollRequest.scroll(scroll);
//                        searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
//                        scrollId = searchResponse.getScrollId();
//                        searchHits = searchResponse.getHits().getHits();
//                        for (SearchHit hit : searchHits) {
//                            tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
//                        }
//                    }
//
//                    // clear scroll id
//                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
//                    clearScrollRequest.addScrollId(scrollId);
//
//                    client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
//
//                    if (tuples.size() == 0) {
//                        continue;
//                    }
//
//                    /**
//                     * 对于获取到的数据,转为GeoItem
//                     */
//
//                    for (Tuple2<String, String> f : tuples) {
//
//                        Gson gson = new Gson();
//                        Map<String, Object> inMap = gson.fromJson(f._2, new HashMap<String, Object>().getClass());
//                        inMap.remove("id");
//                        if (!inMap.containsKey("GEOM")) {
//                            logger.error("Unvalid geometry item for id " + f._1 + "with source " + source + ", must have field named with 'shape'");
//                            continue;
//                        }
//                        Map<String, Object> geoMap = (LinkedTreeMap) inMap.get("GEOM");
//                        String type = (String) geoMap.get("type");
//                        ArrayList t = ((ArrayList) geoMap.get("coordinates"));
//                        GeometryFactory gf = new GeometryFactory();
//                        for (int i = 0; i < t.size(); i++) {
//                            List<List<ArrayList>> coordinates = new ArrayList<>();
//                            ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
//                            if (!type.contains("Multi")) {
//                                coordinates.add(m);
//                            } else {
//                                for (int x = 0; x < m.size(); x++) {
//                                    coordinates.add(m.get(x));
//                                }
//                            }
//                            for (List<ArrayList> coordinate : coordinates) {
//                                Geometry geometry;
//                                if (type.equals("Polygon") || type.equals("MultiPolygon")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createPolygon(cs);
//                                } else if (type.equals("LineString") || type.equals("MultiLineString")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createLineString(cs);
//                                } else {
//                                    // type == Point
//                                    Coordinate c = new Coordinate((double) (coordinate.get(0).get(0)), (double) (coordinate.get(0).get(1)));
//                                    geometry = gf.createPoint(c);
//                                }
//
//                                Map<String, Object> props = new HashMap<>();
//                                if(null == attrs){  //如果没有声明attrs，则默认读取全部属性
//                                    props = inMap;
//                                    props.remove("GEOM");
//                                }
//                                else { //如果声明了attrs，则根据attrs中的key来获取属性
//                                    for (String key : attrs.keySet()) {
//                                        props.put(key, inMap.get(attrs.get(key)));
//                                    }
//                                }
//                                if (tstransformBroad.getValue() != null) {
//                                    geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
//                                }
//
//                                GeoItem gi = new GeoItem();
//                                gi.setFid(f._1);
//                                gi.setProp(props);
//                                gi.setGeometry(geometry);
//
//                                features.add(new Tuple2<>(in._2.toString(), gi));
//                            }
//                        }
//                    }
//
//                    result.addAll(features);
//                }
//                client.close();
//                return result.iterator();
//            }
//        });
        return resultRDD;
    }

    /**
     * 支持以逗号分隔的多个 es source
     *
     * @param jsc
     * @param source
     * @param attrs
     * @param filters
     * @param filterName
     * @param tstransform
     * @return
     */
    public static JavaRDD<GeoItem> readFromES2(JavaSparkContext jsc, String source, Map<String, String> attrs, List<String> filters, String filterName, MathTransform tstransform) throws Exception {
        List<String> sources = Arrays.stream(source.split(",")).map(String::trim).map(x -> x.replace("es://", "")).collect(Collectors.toList());
        List<Tuple2<String, List<String>>> searchQ = new ArrayList<>();
        for (String src : sources) {
            List<String> filterList = new ArrayList<>();
            for (String filter : filters) {
                if (src.contains(filter.substring(0, 2))) {
                    filterList.add(filter);
                }
            }
            if (filterList.size() > 0) {
                searchQ.add(new Tuple2<>(src, filterList));
            }
        }

        Broadcast<MathTransform> tstransformBroad = jsc.broadcast(tstransform);

        JavaRDD<GeoItem> resultRDD = jsc.parallelize(searchQ, (searchQ.size() / 64) + 1).mapPartitions(new FlatMapFunction<Iterator<Tuple2<String, List<String>>>, GeoItem>() {
            @Override
            public Iterator<GeoItem> call(Iterator<Tuple2<String, List<String>>> input) throws Exception {
                String[] ips = ES_IP.split(",");
                HttpHost[] hosts = new HttpHost[ips.length];
                for (int m = 0; m < ips.length; m++) {
                    hosts[m] = new HttpHost(ips[m].trim(), ES_IP_PORT);
                }

                RestHighLevelClient client = new RestHighLevelClient(RestClient.builder((hosts)));
                List<GeoItem> result = new ArrayList<>();

                while (input.hasNext()) {
                    List<GeoItem> features = new ArrayList<>();
                    Tuple2<String, List<String>> in = input.next();
                    List<Tuple2<String, String>> tuples = new ArrayList<>();

                    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
                    //精准匹配
                    sourceBuilder.query(QueryBuilders.termsQuery(filterName, in._2));
                    sourceBuilder.size(ES_SEARCH_SCROLL_SIZE);
                    SearchRequest searchRequest = new SearchRequest();
                    searchRequest.indices(in._1);
                    searchRequest.source(sourceBuilder);
                    searchRequest.scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));

                    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                    SearchHits hits = searchResponse.getHits();
                    SearchHit[] searchHits = hits.getHits();

                    for (SearchHit hit : searchHits) {
                        tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                    }

                    String scrollId = searchResponse.getScrollId();
                    final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
                    while (searchHits != null && searchHits.length > 0) {
                        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                        scrollRequest.scroll(scroll);
                        searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                        scrollId = searchResponse.getScrollId();
                        searchHits = searchResponse.getHits().getHits();
                        for (SearchHit hit : searchHits) {
                            tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                        }
                    }

                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                    clearScrollRequest.addScrollId(scrollId);

                    client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);

                    if (tuples.size() == 0) {
                        continue;
                    }

                    // 对于获取到的数据 转为GeoItem
                    for (Tuple2<String, String> f : tuples) {
                        Gson gson = new Gson();
                        Map<String, Object> inMap = gson.fromJson(f._2, new HashMap<String, Object>().getClass());
                        inMap.remove("id");
                        Map<String, Object> props = new HashMap<>();
                        if(null == attrs){
                            props = inMap;
                            props.remove("GEOM");
                        }
                        else {
                            for (String key : attrs.keySet()) {
                                props.put(key, inMap.get(attrs.get(key)));
                            }
                        }
                        GeoItem gi = new GeoItem();
                        gi.setFid(f._1);
                        gi.setProp(props);
                        features.add(gi);
//                        if (!inMap.containsKey("shape")) {
//                            logger.error("Unvalid geometry item for id " + f._1 + " with source" + source + ",must hae field named with 'shape'");
//                            continue;
//                        }
//                        Map<String, Object> geoMap = (LinkedTreeMap) inMap.get("shape");
//                        String type = (String) geoMap.get("type");
//                        ArrayList t = (ArrayList) geoMap.get("coordinates");
//                        GeometryFactory gf = new GeometryFactory();
//                        for (int i = 0; i < t.size(); i++) {
//                            List<List<ArrayList>> coordinates = new ArrayList<>();
//                            ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
//                            if (!type.contains("Multi")) {
//                                coordinates.add(m);
//                            } else {
//                                for (int x = 0; x < m.size(); x++) {
//                                    coordinates.add(m.get(x));
//                                }
//                            }
//                            for (List<ArrayList> coordinate : coordinates) {
//                                Geometry geometry;
//                                if (type.equals("Polygon") || type.equals("MultiPolygon")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createPolygon(cs);
//                                } else if (type.equals("LineString") || type.equals("MultiLineString")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createLineString(cs);
//                                } else {
//                                    Coordinate c = new Coordinate((double) (coordinate.get(0).get(0)), (double) (coordinate.get(0).get(1)));
//                                    geometry = gf.createPoint();
//                                }
//
//                                Map<String, Object> props = new HashMap<>();
//                                for (String key : attrs.keySet()) {
//                                    props.put(key, inMap.get(attrs.get(key)));
//                                }
//                                if (tstransformBroad.getValue() != null) {
//                                    geometry = JTS.transform(geometry, tstransformBroad.getValue());
//                                }
//
//                                GeoItem gi = new GeoItem();
//                                gi.setFid(f._1);
//                                gi.setProp(props);
//                                gi.setGeometry(geometry);
//
//                                features.add(gi);
//                            }
//                        }
                    }
                    result.addAll(features);
                }
                client.close();
                return result.iterator();
            }
        });
        return resultRDD;
    }

    /**
     * 从ES中读数据，支持以逗号分隔的多个es source.(不读取几何信息)
     * @param jsc
     * @param source
     * @param attrs
     * @param tstransform
     * @return
     * @throws Exception
     */
    public static JavaRDD<GeoItem> readFromES3(JavaSparkContext jsc, String source, Map<String, String> attrs, MathTransform tstransform) throws Exception {
        List<String> sources = Arrays.stream(source.split(",")).map(String::trim).map(x -> x.replace("es://", "")).collect(Collectors.toList());

        Broadcast<MathTransform> tstransformBroad = jsc.broadcast(tstransform);

        JavaRDD<String> result = jsc.parallelize(sources);///how???
        JavaRDD<GeoItem> resultRDD=result.flatMap(new FlatMapFunction<String, GeoItem>() {
            @Override
            public Iterator<GeoItem> call(String input) throws Exception {
                String[] ips = ES_IP.split(",");
                HttpHost[] hosts = new HttpHost[ips.length];
                for(int m = 0; m < ips.length; m++){
                    hosts[m] = new HttpHost(ips[m].trim(), ES_IP_PORT);
                }
                RestHighLevelClient client = new RestHighLevelClient(RestClient.builder((hosts)));
                List<GeoItem> result = new ArrayList<>();


                List<GeoItem> features = new ArrayList<>();
                String in = input;
                List<Tuple2<String, String>> tuples = new ArrayList<>();

                SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
                sourceBuilder.query(QueryBuilders.matchAllQuery());
                sourceBuilder.size(ES_SEARCH_SCROLL_SIZE);
                SearchRequest searchRequest = new SearchRequest();
                searchRequest.indices(in);
                searchRequest.source(sourceBuilder);
                searchRequest.scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));

                SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
                SearchHits hits = searchResponse.getHits();
                SearchHit[] searchHits = hits.getHits();
                for (SearchHit hit : searchHits) {
                    tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                }

                String scrollId = searchResponse.getScrollId();
                final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
                while (searchHits != null && searchHits.length > 0) {
                    SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
                    scrollRequest.scroll(scroll);
                    searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
                    scrollId = searchResponse.getScrollId();
                    searchHits = searchResponse.getHits().getHits();
                    for (SearchHit hit : searchHits) {
                        tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
                    }
                }

                ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
                clearScrollRequest.addScrollId(scrollId);

                client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);



                // 对于获取到的数据 转为GeoItem
                for (Tuple2<String, String> f : tuples) {
                    Gson gson = new Gson();
                    Map<String, Object> inMap = gson.fromJson(f._2, new HashMap<String, Object>().getClass());
                    inMap.remove("id");

                    GeoItem gi = new GeoItem();
                    gi.setFid(f._1);

                    Map<String, Object> geoMap = (LinkedTreeMap) inMap.get("GEOM");
                    String type = (String) geoMap.get("type");
                    ArrayList t = ((ArrayList) geoMap.get("coordinates"));
                    GeometryFactory gf = new GeometryFactory();
                    for (int i = 0; i < t.size(); i++) {
                        List<List<ArrayList>> coordinates = new ArrayList<>();
                        ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
                        if (!type.contains("Multi")) {
                            coordinates.add(m);
                        } else {
                            for (int x = 0; x < m.size(); x++) {
                                coordinates.add(m.get(x));
                            }
                        }
                        for (List<ArrayList> coordinate : coordinates) {
                            Geometry geometry;
                            if (type.equals("Polygon") || type.equals("MultiPolygon")) {
                                Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
                                geometry = gf.createPolygon(cs);
                            } else if (type.equals("LineString") || type.equals("MultiLineString")) {
                                Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
                                geometry = gf.createLineString(cs);
                            } else {
                                // type == Point
                                Coordinate c = new Coordinate((double) (coordinate.get(0).get(0)), (double) (coordinate.get(0).get(1)));
                                geometry = gf.createPoint(c);
                            }
                            if (tstransformBroad.getValue() != null) {
                                geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
                            }
                            gi.setGeometry(geometry);
                        }
                    }

                    Map<String, Object> props = new HashMap<>();
                    if (null == attrs) {
                        props = inMap;
                        props.remove("GEOM");
                    } else {
                        for (String key : attrs.keySet()) {
                            props.put(key, inMap.get(attrs.get(key)));
                        }
                    }
                    gi.setProp(props);
                    features.add(gi);
                }
                result.addAll(features);

                client.close();
                return result.iterator();
            }
        });



//        JavaRDD<GeoItem> resultRDD = result.mapPartitions(new FlatMapFunction<Iterator<String>, GeoItem>() {
//            @Override
//            public Iterator<GeoItem> call(Iterator<String> input) throws Exception {
//                String[] ips = ES_IP.split(",");
//                HttpHost[] hosts = new HttpHost[ips.length];
//                for(int m = 0; m < ips.length; m++){
//                    hosts[m] = new HttpHost(ips[m].trim(), ES_IP_PORT);
//                }
//                RestHighLevelClient client = new RestHighLevelClient(RestClient.builder((hosts)));
//                List<GeoItem> result = new ArrayList<>();
//
//                while(input.hasNext()){
//                    List<GeoItem> features = new ArrayList<>();
//                    String in = input.next();
//                    List<Tuple2<String, String>> tuples = new ArrayList<>();
//
//                    SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
//                    sourceBuilder.query(QueryBuilders.matchAllQuery());
//                    sourceBuilder.size(ES_SEARCH_SCROLL_SIZE);
//                    SearchRequest searchRequest = new SearchRequest();
//                    searchRequest.indices(in);
//                    searchRequest.source(sourceBuilder);
//                    searchRequest.scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
//
//                    SearchResponse searchResponse = client.search(searchRequest, RequestOptions.DEFAULT);
//                    SearchHits hits = searchResponse.getHits();
//                    SearchHit[] searchHits = hits.getHits();
//                    for (SearchHit hit : searchHits) {
//                        tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
//                    }
//
//                    String scrollId = searchResponse.getScrollId();
//                    final Scroll scroll = new Scroll(TimeValue.timeValueMinutes(ES_SEARCH_SCROLL_INTERVAL));
//                    while (searchHits != null && searchHits.length > 0) {
//                        SearchScrollRequest scrollRequest = new SearchScrollRequest(scrollId);
//                        scrollRequest.scroll(scroll);
//                        searchResponse = client.scroll(scrollRequest, RequestOptions.DEFAULT);
//                        scrollId = searchResponse.getScrollId();
//                        searchHits = searchResponse.getHits().getHits();
//                        for (SearchHit hit : searchHits) {
//                            tuples.add(new Tuple2<>(hit.getId(), hit.getSourceAsString()));
//                        }
//                    }
//
//                    ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
//                    clearScrollRequest.addScrollId(scrollId);
//
//                    client.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
//
//                    if (tuples.size() == 0) {
//                        continue;
//                    }
//
//                    // 对于获取到的数据 转为GeoItem
//                    for (Tuple2<String, String> f : tuples) {
//                        Gson gson = new Gson();
//                        Map<String, Object> inMap = gson.fromJson(f._2, new HashMap<String, Object>().getClass());
//                        inMap.remove("id");
//
//                        GeoItem gi = new GeoItem();
//                        gi.setFid(f._1);
//
//                        Map<String, Object> geoMap = (LinkedTreeMap) inMap.get("shape");
//                        String type = (String) geoMap.get("type");
//                        ArrayList t = ((ArrayList) geoMap.get("coordinates"));
//                        GeometryFactory gf = new GeometryFactory();
//                        for (int i = 0; i < t.size(); i++) {
//                            List<List<ArrayList>> coordinates = new ArrayList<>();
//                            ArrayList<ArrayList> m = (ArrayList<ArrayList>) t.get(i);
//                            if (!type.contains("Multi")) {
//                                coordinates.add(m);
//                            } else {
//                                for (int x = 0; x < m.size(); x++) {
//                                    coordinates.add(m.get(x));
//                                }
//                            }
//                            for (List<ArrayList> coordinate : coordinates) {
//                                Geometry geometry;
//                                if (type.equals("Polygon") || type.equals("MultiPolygon")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createPolygon(cs);
//                                } else if (type.equals("LineString") || type.equals("MultiLineString")) {
//                                    Coordinate[] cs = coordinate.stream().map((p) -> new Coordinate((double) (p.get(0)), (double) (p.get(1)))).toArray(Coordinate[]::new);
//                                    geometry = gf.createLineString(cs);
//                                } else {
//                                    // type == Point
//                                    Coordinate c = new Coordinate((double) (coordinate.get(0).get(0)), (double) (coordinate.get(0).get(1)));
//                                    geometry = gf.createPoint(c);
//                                }
//                                if (tstransformBroad.getValue() != null) {
//                                    geometry = JTS.transform(geometry, tstransformBroad.getValue().inverse());
//                                }
//                                gi.setGeometry(geometry);
//                            }
//                        }
//
//                        Map<String, Object> props = new HashMap<>();
//                        if (null == attrs) {
//                            props = inMap;
//                            props.remove("shape");
//                        } else {
//                            for (String key : attrs.keySet()) {
//                                props.put(key, inMap.get(attrs.get(key)));
//                            }
//                        }
//                        gi.setProp(props);
//                        features.add(gi);
//                    }
//                    result.addAll(features);
//                }
//                client.close();
//                return result.iterator();
//            }
//        });
        return resultRDD;
    }

    public static JavaPairRDD<String, GeoItem> mapToPair(JavaRDD<GeoItem> gis, PyramidConfig pc) {
        JavaPairRDD<String, GeoItem> result = gis.flatMapToPair(new PairFlatMapFunction<GeoItem, String, GeoItem>() {
            @Override
            public Iterator<Tuple2<String, GeoItem>> call(GeoItem geoItem) throws Exception {
                List<Tuple2<String, GeoItem>> result = new ArrayList<>();

                Geometry geom = geoItem.getGeometry();
                ReferencedEnvelope envelope = JTS.toEnvelope(geom);
                Double yarea=geom.getArea();
                Double ylength=geom.getLength();
                ZLevelInfo tZLevelInfo = TileUtil.initZLevelInfoPZ(pc, envelope)[1];
                // 针对每一层级的不同瓦片
                int tx_min = tZLevelInfo.getTileRanges()[0];
                int tx_max = tZLevelInfo.getTileRanges()[1];
                int ty_min = tZLevelInfo.getTileRanges()[2];
                int ty_max = tZLevelInfo.getTileRanges()[3];
                int bh=0;
                for (int tile_x = tx_min; tile_x <= tx_max; tile_x++) {
                    for (int tile_y = ty_min; tile_y <= ty_max; tile_y++) {
                        TileID tileID = new TileID();
                        tileID.setX(tile_x);
                        tileID.setY(tile_y);
                        tileID.setzLevel(DEFAULT_Z_LEVEL);
                        Envelope tileEnvelope = TileUtil.createTileBox(tileID, pc);
                        // 得到在每一个瓦片中对应的geometry
                        GeoItem tmpGeoItem = new GeoItem();
                        Geometry tileGeometry = JTS.toGeometry(tileEnvelope);
                        Geometry finalGeom = geom.intersection(tileGeometry);
                        tmpGeoItem.setFid(geoItem.getFid());
                        tmpGeoItem.setGeometry(finalGeom);
                        Map<String,Object> props=geoItem.getProp();
                        props.put("BH",bh);
                        props.put("YTBMJ",yarea);

                        props.put("YLENGTH",ylength);
                        bh++;
                        tmpGeoItem.setProp(props);

                        result.add(new Tuple2<>(tileID.toString(), tmpGeoItem));
                    }
                }
                return result.iterator();
            }
        });
        return result;
    }

}
